package cn.virens.cmpt.server.handler;

import java.net.StandardSocketOptions;
import java.nio.ByteBuffer;
import java.nio.channels.AsynchronousServerSocketChannel;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.CompletionHandler;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.hutool.core.collection.ConcurrentHashSet;
import cn.virens.cmpt.server.ServerManageImpl;
import cn.virens.cmpt.server.channel.Channel;
import cn.virens.cmpt.server.channel.ChannelSimple;
import cn.virens.cmpt.server.listener.OnChannelConnectedListener;
import cn.virens.cmpt.server.listener.OnChannelExceptionListener;
import cn.virens.common.CloseableUtil;

public class AsynchronousSocketAcceptHandler implements CompletionHandler<AsynchronousSocketChannel, Object> {
	private final Logger logger = LoggerFactory.getLogger(AsynchronousSocketAcceptHandler.class);
	private final Map<String, Channel> channelMap = new ConcurrentHashMap<>();
	private final Set<Channel> channelSet = new ConcurrentHashSet<>();

	private final ServerManageImpl serverManage;

	private AsynchronousSocketAcceptHandler(ServerManageImpl serverManage) {
		this.serverManage = serverManage;
	}

	@Override
	public void completed(AsynchronousSocketChannel asynchronousSocketChannel, Object attachment) {
		OnChannelConnectedListener onChannelConnectedListener = serverManage.getOnChannelConnectedListener();
		OnChannelExceptionListener onChannelExceptionListener = serverManage.getOnChannelExceptionListener();
		AsynchronousServerSocketChannel serverSocketChannel = serverManage.getServerSocketChannel();

		try {
			ChannelSimple channel = new ChannelSimple(asynchronousSocketChannel, this);
			channel.setOption(StandardSocketOptions.SO_RCVBUF, 64 * 1024);
			channel.setOption(StandardSocketOptions.SO_SNDBUF, 64 * 1024);
			channel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
			channel.setOption(StandardSocketOptions.SO_KEEPALIVE, true);

			try {
				// 判断连接是否已经打开，如果连接打开就将连接放入连接池
				if (!serverManage.isWaitStop() && channel.isOpen() && channelSet.add(channel)) {
					channel.setDataHandlerListener(serverManage.getDataHandlerListener());
					channel.setDataErrorListener(serverManage.getDataErrorListener());
					channel.setDataWriteListener(serverManage.getDataWriteListener());
					channel.setDataReadListener(serverManage.getDataReadListener());
					channel.setDataDecoder(serverManage.getDataDecoder());
					channel.setDataEncoder(serverManage.getDataEncoder());

					// 开启数据读取监听
					channel.start(ByteBuffer.allocateDirect(10240));

					// 连接已经完成创建，回调连接成功事件
					if (onChannelConnectedListener != null) {
						onChannelConnectedListener.onConnected(channel);
					}
				}
			} catch (Exception ex) {
				// 如果有异常回调监听，就使用异常和连接信息进行回调
				if (onChannelExceptionListener != null) {
					onChannelExceptionListener.onFailed(channel, ex);
				}

				// 从连接池中删掉该连接并安全的关闭该连接
				if (channelSet.remove(channel)) {
					CloseableUtil.close(channel);
				}
			}
		} catch (Exception ex) {
			CloseableUtil.close(asynchronousSocketChannel);
		} finally {

			// 如果还没开始关闭服务，就继续进行等待连接
			if (serverSocketChannel.isOpen() && !serverManage.isWaitStop()) {
				serverSocketChannel.accept(null, this);
			}
		}
	}

	@Override
	public void failed(Throwable exc, Object attachment) {
		try {
			if (serverManage.getOnChannelExceptionListener() != null) {
				serverManage.getOnChannelExceptionListener().onFailed(null, exc);
			}
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		}
	}

	public Channel getChannel(String uid) {
		return channelMap.get(uid);
	}

	public void bindUid(Channel channel, String uid) {
		if (channel instanceof ChannelSimple) {
			((ChannelSimple) channel).setRemove(false);
			((ChannelSimple) channel).setUid(uid);
		}

		// 将用户ID和连接关联起来
		this.channelMap.put(uid, channel);
	}

	public void unbindUid(Channel channel) {
		try {
			// 如果该连接是正常的，并且又我建立的，就调用关闭相关方法
			if (channel instanceof ChannelSimple) {
				((ChannelSimple) channel).setRemove(true);
				((ChannelSimple) channel).close();
			}

			// 如果启动时注册了连接关闭监听，就调用关闭监听
			if (serverManage.getOnChannelCloseableListener() != null) {
				serverManage.getOnChannelCloseableListener().onClose(channel);
			}
		} catch (Exception e) {
			logger.error(e.getMessage(), e);
		} finally {
			this.channelMap.remove(channel.getUid());
			this.channelSet.remove(channel);
		}
	}

	public void execute(Runnable task) {
		this.serverManage.getWokerExecutor().execute(task);
	}

	public static AsynchronousSocketAcceptHandler newInstance(ServerManageImpl serverManage) {
		return new AsynchronousSocketAcceptHandler(serverManage);
	}

}
